package com.ebspp.client.buffer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;


import com.ebspp.client.placement.IPlacement;
import com.ebspp.client.placement.MvDBPlacement;
import com.ebspp.client.serverManager.ReadConnection;
import com.ebspp.client.serverManager.ServerManager;
import com.ebspp.client.serverManager.WriteConnection;
import com.ebspp.global.EBSPPTimer;
import com.ebspp.global.General;
import com.ebspp.thrift.DataMsg;
import com.ebspp.thrift.Identity;
import com.ebspp.thrift.ResourceType;

/**
 * The asynchronous buffer on the client site.
 * It buffers the application read and write data
 * 
 * a buffer pool is related a resource type, which cannot be changed at runtime.
 * if an application wants more than one type of resources, open multiple buffers
 * 
 * @author x23pan
 *
 */
public class MvAsyncBuffer implements IBuffer
{
	private Identity id;
	private int bufID;
	private ResourceType resType;
	
	//buffer size control parameters
	private int maxWriteSize;
	private int maxFailSize;
	private int maxReadSize;
	//record the key->server placement information
	IPlacement placementCB = null;
	
	//application write data buffer
	private Map<String, LinkedBlockingQueue<DataMsg>> serverWQueueMap;
	//those data failed to be stored on the server due to any reasons.
	private LinkedBlockingQueue<DataMsg> wFailQueue;
	
	//application read data buffer
	private Map<String, LinkedBlockingQueue<DataMsg>> serverRQueueMap;
	//data read back from the server
	private LinkedBlockingQueue<DataMsg> rFinQueue;
	
	//those write that are queued in the buffer
	private volatile int writeSize = 0;
	private Map<String, Map<String, DataMsg>> keyMsgMap;
	
	/**
	 * the monitor has the following responsibilities
	 * 1. ask for new connections if the all queues are under pressure and cannot provision more resources
	 * 2. monitor the queue, select a queue for future writes
	 */
	private BufferMonitor monitor = null;
	
	public MvAsyncBuffer(int writeBufSize, int readBufSize, int failListSize, 
						int bufID, Identity id, ResourceType type) throws Exception
	{
		this.id = id;
		this.bufID = bufID;
		this.resType = type;
		this.maxWriteSize = writeBufSize;
		this.maxReadSize = readBufSize;
		this.maxFailSize = failListSize;
		
		//default to use async DB placement mapping, memory placement is also supported
		placementCB = new MvDBPlacement(id, type);
		
		serverWQueueMap = Collections.synchronizedMap(
				new HashMap<String, LinkedBlockingQueue<DataMsg>>());
		wFailQueue = new LinkedBlockingQueue<DataMsg>();
		
		serverRQueueMap = Collections.synchronizedMap(
				new HashMap<String, LinkedBlockingQueue<DataMsg>>());
		rFinQueue = new LinkedBlockingQueue<DataMsg>();
		
		keyMsgMap = Collections.synchronizedMap(new HashMap<String, Map<String, DataMsg>>());
		
		monitor = new BufferMonitor(this);	
	}
	
	//todo
	@Override
	public boolean close() 
	{
		return false;
	}
	
	
	public Identity getID()
	{
		return id;
	}
	@Override
	public int getBufID() 
	{
		return bufID;
	}
	public ResourceType getResType()
	{
		return resType;
	}
	public IPlacement getPlacementCB()
	{
		return placementCB;
	}
	
	public boolean isFullIfWriteSize(int size)
	{
		if(size + writeSize > maxWriteSize)
			return true;
		return false;
	}
	
	/**
	 * writes retrieve the data from the queue, and write to servers.
	 * Don't forget to call cleanBufMsg() after you are sure the data is in the server. 
	 * @param conn
	 * @param size
	 * @return
	 */
	public List<DataMsg> readWQueue(String server, int size)
	{
		if(size < 0)
			size = Integer.MAX_VALUE;
		
		List<DataMsg> data = new ArrayList<DataMsg>();
		serverWQueueMap.get(server).drainTo(data, size);
		
		return data;
	}
	

	/**
	 * read specified key:timestamp from the queue, and read them from the server
	 * @param server
	 * @param size
	 * @return
	 */
	public List<DataMsg> readRQueue(String server, int size)
	{
		if(size < 0)
			size = Integer.MAX_VALUE;
		List<DataMsg> data = new ArrayList<DataMsg>();
		serverRQueueMap.get(server).drainTo(data, size);
		
		return data;
	}
	
	public int getWQueueSize(String server)
	{
		return serverWQueueMap.get(server).size();
	}
	public int getRQueueSize(String server)
	{
		return serverRQueueMap.get(server).size();
	}	
	public LinkedBlockingQueue<DataMsg> getRFinQueue()
	{
		return rFinQueue;
	}
	public LinkedBlockingQueue<DataMsg> getWFinQueue()
	{
		return wFailQueue;
	}
	


	/**
	 * write data to the buffer pool
	 * called by the application
	 * @param msg
	 */
	public boolean write(DataMsg msg)
	{	
		//there might be other choices for selecting a write queue,
		//e.g. server on which the key last stored, minimising garbage collection
		LinkedBlockingQueue<DataMsg> minQueue = monitor.getMinWQueue();
		
		//if the buffer is full, the write will be blocked!
		if(writeSize >= maxWriteSize)
		{
			System.err.println("Waiting for buf space");
			return false;
		}
		
		writeSize++;
		Map<String, DataMsg> tsMsgMap = keyMsgMap.get(msg);
		if(tsMsgMap == null)
		{
			tsMsgMap = Collections.synchronizedMap(new HashMap<String, DataMsg>());
			keyMsgMap.put(msg.getKey(), tsMsgMap);
		}
		tsMsgMap.put(msg.getTs(), msg);
		minQueue.add(msg);
		
		return true;
	}
	/**
	 * clean the buffer after data is written to server
	 * @param data
	 */
	public void cleanBufMsg(List<DataMsg> data)
	{
		for(DataMsg cmsg : data)
		{
			Map<String, DataMsg> tsMsgMap = keyMsgMap.get(cmsg.getKey());
			if(tsMsgMap == null)
			{
				System.err.println("Fail to clean buffer");
				return;
			}
			else if(tsMsgMap.remove(cmsg.getTs()) == null)
				System.err.println("Fail to clean buffer");
			else
				writeSize--;
			
			if(tsMsgMap.size() == 0)
				keyMsgMap.remove(cmsg.getKey());
		}
	}
	
	private DataMsg inBufRead(DataMsg msg)
	{
		if(General.inBufRead)
		{
			Map<String, DataMsg> tsMsgMap = keyMsgMap.get(msg.getKey());
			//note hashmap support null key, which means datamsg without timestamp
			if(tsMsgMap != null)
				return tsMsgMap.get(msg.getTs());
			return null;
		}
		else
			return null;
	}
	
	/**
	 * ask client to read the data from servers
	 * called by the application
	 * for asyncbuffer, it always return null;
	 * @param key
	 */
	public DataMsg read(DataMsg msg)
	{
		//if it is still at the buf, not written to server yet
		DataMsg qMsg = inBufRead(msg);
		if(qMsg != null)
		{
			rFinQueue.add(qMsg);
			return null;
		}
		//find its placement mapping in placementCB
		String server = placementCB.getPlace(msg);
		if(server == null)
		{
			System.err.println("cannot find the data, lost or not exist?: " +
								msg.getKey()+":"+msg.getTs());
			return null;
		}
		
		LinkedBlockingQueue<DataMsg> rQueue = monitor.getRQueue(server);
		rQueue.add(msg);
		
		return null;
	}


	public void setWFail(List<DataMsg> failList)
	{
		if(failList == null)
			return;
		
		while(failList.size() + wFailQueue.size() > maxFailSize)
		{
			DataMsg msg = wFailQueue.poll();
			System.err.println("Fail write queue drop: " + msg.getKey()+"->"+msg.getTs());
		}
		wFailQueue.addAll(failList);
	}
	
	/**
	 * append finished reads.
	 * Finished keys are the keys that are stored by the server.
	 * If read buffer is full, it will block all following reads
	 * 
	 * Called by the connection monitor
	 * @param finList
	 */
	public void setRFin(List<DataMsg> finList)
	{
		if(finList == null)
			return;
		
		while(finList.size() + rFinQueue.size() > maxReadSize)
		{
			System.err.println("Waiting for read buffer");
			
			try {
				Thread.sleep(EBSPPTimer.readBufWaitT);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		rFinQueue.addAll(finList);
	}
	
	
	/**
	 * read what data is failed in the write process
	 * called by the application
	 * @param size
	 * @return
	 */
	public List<DataMsg> readWFail(int size)
	{
		List<DataMsg> finList = new ArrayList<DataMsg>();
		if(size < 0)
			size = Integer.MAX_VALUE;
		
		wFailQueue.drainTo(finList, size);
		return finList;	
	}
	
	/**
	 * read the data which is already read back from the server
	 * called by application
	 * @param size
	 * @return
	 */
	public List<DataMsg> readRFin(int size)
	{
		List<DataMsg> dataList = new ArrayList<DataMsg>();
		if(size < 0)
			size = Integer.MAX_VALUE;
		
		rFinQueue.drainTo(dataList, size);
		return dataList;
	}
	

	
	/**
	 * currently the buffer monitor has two responsibilities
	 * 1. if all connections have use up their resources and cannot provision any more.
	 * it opens another "good" connection by consulting the server manager
	 * 2. it tries to find a connection for the next write to server. Currently it uses the shortest
	 * queue as the candidate
	 * 
	 * @author x23pan
	 *
	 */
	class BufferMonitor
	{
		MvAsyncBuffer buf;
		private LinkedBlockingQueue<DataMsg> minWQueue = null;
		
		public BufferMonitor(MvAsyncBuffer buf)
		{
			this.buf = buf;
			
			//create the first connection
			WriteConnection wConn = 
					ServerManager.getWCandidate(resType, buf.getBufID(), buf.getID());
			createWriter(wConn.getServer());
			minWQueue = serverWQueueMap.get(wConn.getServer());
			
			//create write monitor
			wQueueMonitor();
		}
		
		
		private void createWriter(String server)
		{
			LinkedBlockingQueue<DataMsg> wQueue = new LinkedBlockingQueue<DataMsg>();
			serverWQueueMap.put(server, wQueue);
			
			MvAsyncBufWriter writer = new MvAsyncBufWriter(buf, server);
			Thread writerT = new Thread(writer, "AsyncWriter:" + server);
			writerT.start();
			
			System.out.println("Add a write connection to: " + server);
			System.out.println("Total write connection: " + serverWQueueMap.size());
		}
		
		/**
		 * monitor the write queues
		 */
		private void wQueueMonitor()
		{
			new Thread(new Runnable()
			{
				@Override
				public void run() 
				{
					// TODO Auto-generated method stub
					while(true)
					{
						try {
							Thread.sleep(EBSPPTimer.newConnectionManagerT);
						} catch (InterruptedException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
						//analyze the write queues.
						analyzeWQueue();
					}
				}
			}, "Write Monitor").start();
		}
		
		/**
		 * If a connection cannot provision more resource, its resource resource value will become negative.
		 * Hence, if all resources are negative, the monitor open another connection by consulting the
		 * ServerManager
		 */
		private void analyzeWQueue()
		{
			boolean add = true;
			int minSize = Integer.MAX_VALUE;
			LinkedBlockingQueue<DataMsg> tmpMin = null;
			
			Iterator<String> serverIt = serverWQueueMap.keySet().iterator();
			synchronized(serverWQueueMap)
			{
				while(serverIt.hasNext())
				{
					String server = serverIt.next();
					
					if(ServerManager.getWConnRes(server, bufID) > 0)
						add = false;
					
					LinkedBlockingQueue<DataMsg> wQueue = serverWQueueMap.get(server);
					if(wQueue.size() < minSize)
					{
						minSize = wQueue.size();
						tmpMin = wQueue;
					}
				}
			}
			if(add)
			{
				WriteConnection conn = 
						ServerManager.getWCandidate(resType, buf.getBufID(), buf.getID());
				createWriter(conn.getServer());
			}
			minWQueue = tmpMin;
		}
		
		/**
		 * get the queue that has potential minimal latency
		 * @return
		 */
		public LinkedBlockingQueue<DataMsg> getMinWQueue()
		{
			return minWQueue;
		}
		
		/**
		 * get a read connection to a specific server
		 * if not exists, create one.
		 * @param server
		 * @return
		 */
		public LinkedBlockingQueue<DataMsg> getRQueue(String server)
		{
			LinkedBlockingQueue<DataMsg> rQueue = null;
			if(serverRQueueMap.get(server) == null)
			{
				ReadConnection rConn = ServerManager.createRConn(server, resType, bufID, id);
				if(rConn == null)
					return null;
				
				rQueue = new LinkedBlockingQueue<DataMsg>();
				serverRQueueMap.put(server, rQueue);
				
				MvAsyncBufReader reader = new MvAsyncBufReader(buf, server);
				Thread readerT = new Thread(reader, "AsyncReader:" + server);
				readerT.start();
				System.out.println("Add a read connection to: " + server);
				System.out.println("Total read connection: " + serverRQueueMap.size());
			}
			else
				rQueue = serverRQueueMap.get(server);
			
			return rQueue;
		}
	}



}


